/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.mem.impl;

import java.nio.BufferOverflowException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.sabayframework.mem.BigBuffer;

/**
 * {@link BigBufferAllocator} implementation that instantiate {@link BigBuffer}s of fixed size, called slices.
 */
public class FixedSizeBigBufferAllocator extends AbstractBigBufferAllocator
    
{

    // Collection that keeps track of the parent buffers (segments) where slices are allocated
    private final List<BigBuffer> segmentsBuffers;

    // Collection that owns all slices that can be used.
    private final Queue<BigBuffer> freeBuffers = new ConcurrentLinkedQueue<BigBuffer>();

    // Size of each slices dividing each segments of the slab
    private final long sliceSize;

    // Total size of the current slab
    private long totalSize;

    // Tells if it returns null or throw an BufferOverflowException when the requested size is bigger than the size of the slices
    private boolean returnNullWhenOversizingSliceSize = true;

    // Tells if it returns null when no buffers are available
    private boolean returnNullWhenNoBufferAvailable = true;

    // Collection that keeps track of borrowed buffers
    private final Map<Integer, BigBuffer> usedSliceBuffers = new ConcurrentHashMap<Integer, BigBuffer>();


    /**
     * Constructor.
     *
     * @param number           : internal identifier of the allocator
     * @param totalSize        : the internal buffer
     * @param sliceSize        : arbitrary number of the buffer.
     * @param numberOfSegments : number of parent {@link BigBuffer} to allocate.
     */
    public FixedSizeBigBufferAllocator( final long totalSize, final long sliceSize,
                                             final int numberOfSegments )
    {
        this.totalSize = totalSize;
        this.sliceSize = sliceSize;

        this.segmentsBuffers = new ArrayList<BigBuffer>( numberOfSegments );

        init( numberOfSegments );

    }

    protected void init( final int numberOfSegments )
    {
        if( numberOfSegments <= 0 )throw new IllegalArgumentException("number of segments must be > 0: "+numberOfSegments);

        // Compute the size of each segments
        long segmentSize = totalSize / numberOfSegments;
        // size is rounded down to a multiple of the slice size
        segmentSize -= segmentSize % sliceSize;

        for ( int i = 0; i < numberOfSegments; i++ )
        {
            final BigBuffer segment = new UnsafeDirectBuffer( segmentSize );
            segmentsBuffers.add( segment );

            for ( int j = 0; j < segment.capacity(); j += sliceSize )
            {
                segment.clear();
                segment.position( j );
                segment.limit( j + sliceSize );
                final BigBuffer slice = segment.slice();
                freeBuffers.add( slice );
            }
        }
    }


    protected BigBuffer findFreeBuffer( long capacity )
    {
        // ensure the requested size is not bigger than the slices' size
        if ( capacity > sliceSize )
        {
            if ( returnNullWhenOversizingSliceSize )
            {
                return null;
            }
            else
            {
                throw new BufferOverflowException();
            }
        }
        // TODO : Add capacity to wait till a given timeout for a freed buffer
        return freeBuffers.poll();
    }

    
    @Override
    public void free( final BigBuffer buffer )
    {

        if( isClosed() ) throw new IllegalStateException("Allocator is closed");

        if(buffer == EmptyBigBuffer.INSTANCE) return;
        if ( usedSliceBuffers.remove( getHash( buffer ) ) == null )
        {
            return;
        }

        // Ensure the buffer belongs to this slab
        if( buffer.capacity() != sliceSize ) throw new IllegalArgumentException("Given BigBuffer was allocated by this allocator");

        freeBuffers.offer( buffer );

    }

    
    @Override
    public BigBuffer allocate( long size )
    {

        if( isClosed() ) throw new IllegalStateException("Allocator is closed");

        if(size == 0) return EmptyBigBuffer.INSTANCE;
        BigBuffer allocatedBigBuffer = findFreeBuffer( size );

        if ( allocatedBigBuffer == null )
        {
            if ( returnNullWhenNoBufferAvailable )
            {
                return null;
            }
            else
            {
                throw new BufferOverflowException();
            }
        }

        // Reset buffer's state
        allocatedBigBuffer.clear();
        allocatedBigBuffer.limit( size );

        usedSliceBuffers.put( getHash( allocatedBigBuffer ), allocatedBigBuffer );

        return allocatedBigBuffer;

    }

    public long getSliceSize()
    {
        return sliceSize;
    }

    
    @Override
    public void clear()
    {
        for ( final Map.Entry<Integer, BigBuffer> entry : usedSliceBuffers.entrySet() )
        {
            freeBuffers.offer( entry.getValue() );
        }
        usedSliceBuffers.clear();
    }

    
    @Override
    public long getCapacity()
    {
        return totalSize;
    }

    @Override
    public void close()
    {
        if( isClosed() ) return;

        setClosed( true );

        clear();

        for ( final BigBuffer buffer : segmentsBuffers )
        {
            try
            {
            	buffer.destroy();
            }
            catch ( Exception e )
            {
            }
        }
    }
}